package com.bridgeintelligent.tag.bulkload.service.core.append;

import com.bridgeintelligent.tag.bulkload.api.BulkLoadConfig;
import com.bridgeintelligent.tag.bulkload.service.core.threadpool.ThreadPoolExecutor;
import com.bridgeintelligent.tag.mapper.EsIndexNameMapper;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Service;

import java.io.*;
import java.util.*;

@Slf4j
@AllArgsConstructor
@Service
public class DimAppendHandler {

    private EsIndexNameMapper esIndexNameMapper;

    private ThreadPoolExecutor threadPoolExecutor;

    private static final String SUB_SIGN = "|";

    public void handler(DimAppendMqMsg dimAppendMqMsg) {
        try {
            //根据主题id获取索引名
            String indexName = esIndexNameMapper.findIndexName("1");
            String filePath = dimAppendMqMsg.getFilePath();
            if (!new File(filePath).exists()) {
                log.error("file not exists:" + filePath);
                return ;
            }
            Set<String> set = new HashSet<>();
            List<String> list = new ArrayList<>();
            BufferedReader br;
            br = new BufferedReader(new InputStreamReader(new FileInputStream(filePath)));
            String s;
            while ((s = br.readLine()) != null) {
                if (set.size() == BulkLoadConfig.ELEMENT_COUNT){
                    list = new ArrayList<>(set);
                    loadHandler(list, dimAppendMqMsg, indexName);
                    set = new HashSet<>();
                }
                String value = s.replaceAll("\uFEFF", "");
                set.add(value);
            }
            if (set.size() > 0){
                list = new ArrayList<>(set);
                loadHandler(list, dimAppendMqMsg, indexName);
            }
        } catch (IOException e) {
            e.printStackTrace();
            log.error("\n历史数据追加新标签处理异常----------", e);
            throw new RuntimeException(e);
        }
    }

    public void loadHandler(List<String> list, DimAppendMqMsg dimAppendMqMsg, String indexName){
        List<Map<String, Map<String, Object>>> mapList = list(list, dimAppendMqMsg.getDimId());
        threadPoolExecutor.handler(indexName, dimAppendMqMsg.getIndexType(), mapList, null, dimAppendMqMsg.getType());
    }


    List<Map<String, Map<String, Object>>> list(List<String> list, String dimId) {
        int size = list.size();
        List<Map<String, Map<String, Object>>> result = new ArrayList<>();
        for (int i = 1; i <= BulkLoadConfig.THREAD_COUNT; i++) {
            Map<String, Map<String, Object>> map = new HashMap<>(size / BulkLoadConfig.THREAD_COUNT);
            int start = size / BulkLoadConfig.THREAD_COUNT * (i - 1);
            int end = size * i / BulkLoadConfig.THREAD_COUNT;
            for (int j = start; j < end; j++) {
                Map<String, Object> dimValueMap = new HashMap<>(16);
                dimValueMap.put(dimId, StringUtils.substringAfter(list.get(j), SUB_SIGN));
                map.put(StringUtils.substringBefore(list.get(j), SUB_SIGN), dimValueMap);
            }
            result.add(map);
        }
        return result;
    }

}
